package com.wantup.springboot.boot.jms.mkw.rocketmq.zhucong;

import com.wantup.springboot.boot.jms.mkw.rocketmq.qurickstart.constants.Const;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

/**
 * @author Zhaoluowei
 * @version 1.0
 * @date 2018/11/19 17:15
 * 从节点  要有主节点的副本
 */
public class Consumer {

    //消费者运行的方法
    public static void main(String[] args) throws MQClientException {
        //最常用的是push 也需要先创建一个组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_quick_consumer_name");
        consumer.setNamesrvAddr(Const.NAME_SERVER_ADDR_MASTER_SLAVE);
        //设置位置
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        //要做一个订阅：  第二个参数 指定消费的TagA 也可以写* 代表只要是这个消息队列的都消费
        consumer.subscribe("test_quick_topic2", "*");
        consumer.setVipChannelEnabled(false);

        //设置监听
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //单条消息直接get(0)
                MessageExt messageExt = list.get(0);
                try {
                    String topic = messageExt.getTopic();
                    String tags = messageExt.getTags();
                    String keys = messageExt.getKeys();
//                    if (keys.equals("key-1")){
//                        System.out.println("消费失败。。。。");
//                        int a = 1/0;
//                    }

                    String body = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);

                    System.err.println("接收到的消息: topic: " + topic + " -- tags " + tags + " -- keys " + keys + " -- body " + body);
                } catch (Exception ex) {
                    ex.printStackTrace();
                    //过一段时间重试
//                    int reconsumeTimes = messageExt.getReconsumeTimes();
//                    System.out.println("reconsumeTimes : " + reconsumeTimes);
//                    if (reconsumeTimes == 3) {
//                        //记录日志
//                        //补偿处理
//                        System.out.println("3次了  还是失败  我要 记录日志 补偿处理");
//                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                //消费成功要给个返回值
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.err.println("consumer.start...");
    }

}
